package com.book.kafka.client;

import com.book.kafka.client.record.JsonSerializer;
import com.book.kafka.client.record.KafkaRecord;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
/**
 * Description：
 * @Author： leo.xiong
 * @CreateDate： 2019/10/12 17:14
 * @Email： leo.xiong@suyun360.com
 * Version：1.0
 */
@Slf4j
@Component
public class KafkaClient {

    @Autowired
    private KafkaTemplate<String, byte[]> kafkaTemplate;

    public <T> void send(String topic, T data) {
        byte[] bytes = JsonSerializer.serialize(data);
        this.send(topic, bytes);
    }

    public void send(String topic, Integer partition, Long timestamp, String key, byte[] bytes) {
        kafkaTemplate.send(topic, partition, timestamp, key, bytes);
        log.info("Send data to kafka, topic = " + topic);
    }

    public <T> void send(String topic, Integer partition, Long timestamp, String key, T data) {
        byte[] bytes = JsonSerializer.serialize(data);
        this.send(topic, partition, timestamp, key, bytes);
    }

    public void send(String topic, byte[] bytes) {
        kafkaTemplate.send(topic, bytes);
        log.info("Send data to kafka, topic = " + topic);
    }

    /**
     * It will wrap up data to a KafkaRecord<T>, then send
     *
     * @param topic topic
     * @param key   partition key
     * @param data  payload data
     */
    public <T> void sendWithWrap(String topic, String key, T data) {
        KafkaRecord<T> kafkaRecord = new KafkaRecord<>(topic, key, data);
        this.send(topic, null, null, key, kafkaRecord);
    }


    public <T> void send(String topic, String key, T data) {
        this.send(topic, null, null, key, data);
    }

}
